package net.hn.hnms.biz.upkafka.service;

import net.hn.hnms.biz.upkafka.config.KafkaProperties;
import net.hn.hnms.biz.upkafka.config.SpringContextUtil;
import net.hn.hnms.biz.upkafka.domain.CoalMineSysPointPo;
import net.hn.hnms.biz.upkafka.domain.FanrealNowPo;
import net.hn.hnms.biz.upkafka.mapper.CoalMineSysPointMapper;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class ZTFProducer extends Thread {
    private static final Logger LOG = LoggerFactory.getLogger(Producer.class);

    public final KafkaProducer<String, String> producer;

    private final String topic;
    private final String mineCode;

    public final Boolean isAsync;
    public ZTFProducer(String topicName, Boolean asyncEnable,String mineCode) {
        Properties props = Producer.initProperties(topicName,mineCode);
        producer = new KafkaProducer<String, String>(props);
        topic = topicName;
        isAsync = asyncEnable;
        this.mineCode = mineCode;
    }
    @Override
    public void run() {
        // 构造消息记录
        while (true) {
            StringBuffer message = null;
            if ("150621011015".equals(mineCode) || "150781013671".equals(mineCode)) {

                //高头窑主通风需要特殊处理
                //灵露需要特殊处理
                 message = this.getZTFSSMessage(this.mineCode);
            } else {
                //其他煤矿查询所有测点
                message = this.getZTFSSALLMessage(this.mineCode);
            }

            long startTime = System.currentTimeMillis();
            String key = String.valueOf(mineCode);
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, key, message.toString());
            if (isAsync) {
                // 异步发送
                producer.send(record);
            } else {
                try {
                    // 同步发送
                    producer.send(record).get();
                    System.out.println("发送成功:"+message);
                } catch (InterruptedException ie) {
                    LOG.info("The InterruptedException occured : {}.", ie);
                } catch (ExecutionException ee) {
                    LOG.info("The ExecutionException occured : {}.", ee);
                }
            }
            try {
                // 线程阻塞 1 分钟 (60000 毫秒)
                Thread.sleep(60000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
    private StringBuffer getZTFSSMessage(String mineCode) {
        String mineName = KafkaProperties.getMineNameByMineCode(mineCode);
        String systemName = "主通风监控系统";
        String systemCode = "ZTF";
        String statusCode="30007";  //测点类型编码-运行状态
        String kg = "KG";
        String deviceCode1 = "000000000001";
        String deviceCode2 = "000000000002";
        String deviceCodeOther1 = "100000000001";
        String deviceCodeOther2 = "100000000002";
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        //数据上传时间
        LocalDateTime now = LocalDateTime.now();
        // Define the desired format
        DateTimeFormatter uploadedTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        // Format the current date and time
        //数据上传时间
        String uploadedTime = now.format(uploadedTimeFormatter);
        //构建消息内容
        StringBuffer message = new StringBuffer();
        //主通风系统接入频率：每60s上传一次测点实时数据。
        //构建消息头：包括煤矿编码，煤矿名称，系统型号，系统名称，数据上传时间
        String messageHeader = mineCode + ";"+mineName+";;"+systemName+";" + simpleDateFormat.format(new Date()) + "~";
        message.append(messageHeader);

        StringBuffer messageBody = new StringBuffer();
        //构建主通风消息体：测点编码，测点类型编码，测点类型名称，所属设备编码，测点数值，测点数值单位，测点状态，数据时间~

        //获取 主通风数据
        SystemService systemService = SpringContextUtil.getBean(SystemService.class);
        //获取主通风系统数据
        Map<String, FanrealNowPo> fanrealNowPoMap = systemService.getZtfSystemStatus(mineCode);
        //数据不为空
        if(fanrealNowPoMap!=null && fanrealNowPoMap.size()>0){
            String fanStart1="";
            String fanStart2="";

            String fanOther1 = "";
            String fanOther2 = "";

            //特殊处理运行状态
            String runningStatus01 = "0";
            String runningStatus02 = "0";

            if ("150781013671".equals(mineCode)) {
                //灵露
                fanStart1 = "NewSite_主通风_主通风2_一号风机运行_1";
                fanStart2 = "NewSite_主通风_主通风1_二号风机运行";


                FanrealNowPo a1 = fanrealNowPoMap.get("NewSite_主通风_主通风1_一号变频运行反馈");
                FanrealNowPo a2 = fanrealNowPoMap.get("NewSite_主通风_主通风1_二号变频运行反馈");
                FanrealNowPo a3 = fanrealNowPoMap.get("NewSite_主通风_主通风2_一号变频运行反馈_1");
                FanrealNowPo a4 = fanrealNowPoMap.get("NewSite_主通风_主通风2_二号变频运行反馈_1");
                if (1 == a1.getPointValue().intValue() || 1 == a2.getPointValue()) {
                    runningStatus01 = "1";
                }
                if (1 == a3.getPointValue().intValue() || 1 == a4.getPointValue()) {
                    runningStatus02 = "1";
                }

            }else if ("620824009734".equals(mineCode)) {
                //研北
                fanStart1 = "";
                fanStart2 = "";
            }else if ("150621011015".equals(mineCode)) {
                //高头窑
                //赵家渠
//                fanStart1 = "ST08_BP11_YX";
//                fanStart2 = "ST08_BP21_YX";

                //白家渠
                fanOther1 = "ST24_BP11_YX";
                fanOther2 = "ST24_BP21_YX";
            }
            //主通风1系统启动
            FanrealNowPo po1 = fanrealNowPoMap.get(fanStart1);
            if (po1 != null) {
                FanrealNowPo vo1 = new FanrealNowPo();
                vo1.setDataTime(po1.getDataTime() != null ? po1.getDataTime() : new Date());
                vo1.setPointValue(po1.getPointValue() != null ? po1.getPointValue() : 0);
                vo1.setPointStatus(po1.getPointStatus() != null ? po1.getPointStatus() : 1);
                if ("150781013671".equals(mineCode)) {
                    messageBody.append(mineCode + systemCode + statusCode + kg + deviceCode1 + ";" + statusCode + ";运行状态;" + systemCode + "01;" + runningStatus01 + ";;" + vo1.getPointStatus() + ";" + simpleDateFormat.format(vo1.getDataTime()) + "~");
                } else {
                    messageBody.append(mineCode+systemCode+statusCode+kg+deviceCode1+";"+statusCode+";运行状态;"+systemCode+"01;" + vo1.getPointValue().intValue()+ ";;" + vo1.getPointStatus() + ";" +simpleDateFormat.format(vo1.getDataTime()) + "~");
                }
            }
            //主通风2系统启动
            FanrealNowPo po2 = fanrealNowPoMap.get(fanStart2);
            if (po2 != null) {
                FanrealNowPo vo2 = new FanrealNowPo();
                vo2.setDataTime(po2.getDataTime() != null ? po2.getDataTime() : new Date());
                vo2.setPointValue(po2.getPointValue() != null ? po2.getPointValue() : 0);
                vo2.setPointStatus(po2.getPointStatus() != null ? po2.getPointStatus() : 1);
                if ("150781013671".equals(mineCode)) {
                    messageBody.append(mineCode + systemCode + statusCode + kg + deviceCode2 + ";" + statusCode + ";运行状态;" + systemCode + "02;" + runningStatus02 + ";;" + vo2.getPointStatus() + ";" + simpleDateFormat.format(vo2.getDataTime()) + "~");
                } else {
                    messageBody.append(mineCode + systemCode + statusCode + kg + deviceCode2 + ";" + statusCode + ";运行状态;" + systemCode + "02;" + vo2.getPointValue().intValue() + ";;" + vo2.getPointStatus() + ";" + simpleDateFormat.format(vo2.getDataTime()) + "~");
                }
            }

            //主通风1系统启动-其他
            FanrealNowPo poOther1 = fanrealNowPoMap.get(fanOther1);
            if (poOther1 != null) {
                FanrealNowPo voOther1 = new FanrealNowPo();
                voOther1.setDataTime(poOther1.getDataTime() != null ? poOther1.getDataTime() : new Date());
                voOther1.setPointValue(poOther1.getPointValue() != null ? poOther1.getPointValue() : 0);
                voOther1.setPointStatus(poOther1.getPointStatus() != null ? poOther1.getPointStatus() : 1);
                messageBody.append(mineCode+systemCode+statusCode+kg+deviceCodeOther1+";"+statusCode+";运行状态;"+systemCode+"01;" + voOther1.getPointValue().intValue()+ ";;" + voOther1.getPointStatus() + ";" +simpleDateFormat.format(voOther1.getDataTime()) + "~");
            }
            //主通风2系统启动-其他
            FanrealNowPo poOther2 = fanrealNowPoMap.get(fanOther2);
            if (poOther2 != null) {
                FanrealNowPo voOther2 = new FanrealNowPo();
                voOther2.setDataTime(poOther2.getDataTime() != null ? poOther2.getDataTime() : new Date());
                voOther2.setPointValue(poOther2.getPointValue() != null ? poOther2.getPointValue() : 0);
                voOther2.setPointStatus(poOther2.getPointStatus() != null ? poOther2.getPointStatus() : 1);
                messageBody.append(mineCode+systemCode+statusCode+kg+deviceCodeOther2+";"+statusCode+";运行状态;"+systemCode+"02;" + voOther2.getPointValue().intValue()+ ";;" + voOther2.getPointStatus() + ";" + simpleDateFormat.format(voOther2.getDataTime()) + "~");
            }


        }


        //构建消息结尾
        messageBody.append("||");
        message.append(messageBody);
        return message;
    }
    public StringBuffer getZTFSSALLMessage(String mineCode) {
        String mineName = KafkaProperties.getMineNameByMineCode(mineCode);
        String systemName = "主通风监控系统";
        String systemCode = "ZTF";

        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        //数据上传时间
        LocalDateTime now = LocalDateTime.now();
        // Define the desired format
        DateTimeFormatter uploadedTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        // Format the current date and time
        //数据上传时间
        String uploadedTime = now.format(uploadedTimeFormatter);
        //构建消息内容
        StringBuffer message = new StringBuffer();

        //构建消息头：包括煤矿编码，煤矿名称，系统型号，系统名称，数据上传时间
        String messageHeader = mineCode + ";"+mineName+";"+systemCode+";"+systemName+";" + simpleDateFormat.format(new Date()) + "~";
        message.append(messageHeader);

        //构建主通风消息体：测点编码，测点类型编码，测点类型名称，所属设备编码，测点数值，测点数值单位，测点状态，数据时间~
        StringBuffer messageBody = new StringBuffer();


        //获取 主通风数据
        CoalMineSysPointMapper coalMineSysPointMapper = SpringContextUtil.getBean(CoalMineSysPointMapper.class);
        //获取主通风系统数据
        if ("150621011015".equals(mineCode)) {
            //高头窑的煤矿编码需要特殊处理
            mineCode = "150621011038";
        }
        List<CoalMineSysPointPo> ztFystemList = coalMineSysPointMapper.getZTFystemList(mineCode);


        if ("150621011038".equals(mineCode)) {
            //需要转义回来
            mineCode = "150621011015";
        }

        //数据不为空
        if (!ztFystemList.isEmpty()) {
            ztFystemList.forEach(po -> {
                //判断当前数据是MN 还是KG
                FanrealNowPo vo = new FanrealNowPo();
                //测点编码
                vo.setPointCode(po.getPointCode()!= null ? po.getPointCode() : "");
                //测点类型编码
                vo.setPointTypeCode(po.getSensorTypeCode() != null ? po.getSensorTypeCode() : "");
                //测点类型名称
                vo.setPointTypeName(po.getSensorTypeName() != null ? po.getSensorTypeName() : "");
                //所属设备编码
                vo.setDeviceCode(po.getDeviceArrayIndex() != null ? po.getDeviceArrayIndex() : "");
                //数据时间
                vo.setDataTime(po.getDataTime() != null ? po.getDataTime() : new Date());
                //测点值
                vo.setPointValue(po.getPointCurrentValue() != null ? po.getPointCurrentValue() : 0);
                //值单位
                vo.setPointValueUnit(po.getSensorUnit() != null ? po.getSensorUnit() : "");
                //测点状态
                vo.setPointStatus(po.getSensorStatus() != null ? po.getSensorStatus() : 1);
                // 使用三元表达式检查 valueType
                String formattedValue;
                if ("KG".equals(po.getValueType())) {
                    // 如果是 KG，则取整
                    formattedValue = String.valueOf(vo.getPointValue() != null ? vo.getPointValue().intValue() : 0);
                } else {
                    // 否则保留浮点数格式
                    formattedValue = String.valueOf(vo.getPointValue() != null ? vo.getPointValue() : 0.0f);
                }
                //构建主通风消息体：测点编码，测点类型编码，测点类型名称，所属设备编码，测点数值，测点数值单位，测点状态，数据时间~
                messageBody.append(vo.getPointCode()).append(";")
                        .append(vo.getPointTypeCode()).append(";")
                        .append(vo.getPointTypeName()).append(";")
                        .append(systemCode).append("0").append(vo.getDeviceCode()).append(";")
                        .append(formattedValue).append(";")
                        .append(vo.getPointValueUnit()).append(";")
                        .append(vo.getPointStatus()).append(";")
                        .append(simpleDateFormat.format(vo.getDataTime())).append("~");
            });
        }


        //构建消息结尾
        messageBody.append("||");
        message.append(messageBody);
        return message;
    }
}
